package com.nami.pchat.controller.middleware.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import com.google.protobuf.InvalidProtocolBufferException;
import com.nami.gate.external.basic.bo.ConnectedUserBO;
import com.nami.gate.external.basic.constants.UpTopicConst;
import com.nami.im.network.model.message.v2.Message;
import com.nami.pchat.c2c.PrivateChatFacade;
import com.nami.pchat.c2c.request.PrivateChatCmd;
import com.nami.transport.proto.codec.ImMessages;
import com.nami.transport.proto.codec.MqTransMsg;
import com.nami.transport.utils.MsgTimeUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.json.utils.JsonUtils;
import org.springframework.stereotype.Component;


@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.C2C_TOPIC,
    consumerGroup = UpTopicConst.C2C_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class PrivateMsgConsumer implements RocketMQListener<MessageExt> {

    private final PrivateChatFacade privateChatFacade;

    @Override
    public void onMessage(MessageExt msg) {
        try {
            TimeInterval timer = DateUtil.timer();
            // 1. 解析
            PrivateChatCmd cmd = parseGateMsgBody(msg);
            // 2. 检查消息时效性
            if (cmd == null || MsgTimeUtil.isMessageExpired(cmd.getTimestamp())) {
                log.warn("[pchat] 消息过期，不予处理. mq消息消费成功！留存param={}", JsonUtils.toJsonString(cmd));
                return;
            }
            privateChatFacade.execute(cmd);
        } catch (Exception var3) {
            var3.printStackTrace();
        }

    }

    private PrivateChatCmd parseGateMsgBody(MessageExt msg) {
        MqTransMsg.Router2LogicPublishMsg publishMsg;
        ImMessages.UpMsg upMsg;
        try {
            publishMsg = MqTransMsg.Router2LogicPublishMsg.parseFrom(msg.getBody());
            upMsg  = ImMessages.UpMsg.parseFrom(publishMsg.getData());
        } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException(e);
        }

        MqTransMsg.TransUser sender = publishMsg.getSender();

        Message message = new Message();
        message.setFlags(upMsg.getFlags());
        message.setMsgType(upMsg.getMsgType());
        message.setMsgContent(upMsg.getMsgContent().toStringUtf8());
        message.setMsgTime(upMsg.getMsgTime());
        message.setClientUid(upMsg.getClientUid());
        message.setChannelType(ImMessages.ChannelType.Private_VALUE);
        message.setTargetId(publishMsg.getTargetId());
        message.setSenderId(sender.getUserId());

        PrivateChatCmd cmd = new PrivateChatCmd();
        cmd.setQos(publishMsg.getQos());
        cmd.setMessage(message);
        cmd.setReceiverId(publishMsg.getTargetId());
        cmd.setTimestamp(publishMsg.getTimestamp());

        ConnectedUserBO user = new ConnectedUserBO();
        user.setUserId(sender.getUserId());
        user.setPlatform(sender.getPlatform());
        user.setAppKey(sender.getAppKey());
        user.setConnectId(sender.getConnectId());
        user.setInstanceId(sender.getInstanceId());
        user.setUserIp(sender.getUserIp());
        cmd.setSender(user);

        return cmd;
    }

}
